package com.cl.ks.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.cl.ks.config.ValueConfig;
import com.cl.ks.dto.PileDataClue;
import com.cl.ks.entity.*;
import com.cl.ks.repository.KsFilterRepository;
import com.cl.ks.repository.KsFilterResultRepository;
import com.cl.ks.utils.DbUtil;
import com.cl.ks.utils.SqlConvert;
import com.cl.ks.utils.TaskServiceUtil;
import com.cl.pile.rule.process.RuleMatcher;
import com.cl.ks.constant.FieldConstants;
import com.querydsl.core.types.dsl.BooleanExpression;
import com.querydsl.jpa.impl.JPAQueryFactory;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.java.Log;
import okhttp3.*;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.annotation.Lazy;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import static com.cl.ks.constant.QTableConstant.*;
import static com.cl.ks.utils.SqlConvert.AND;
import static com.cl.ks.utils.SqlConvert.IN;


//@Service
@Service
@Log
@EnableAsync(proxyTargetClass = true)
public class FilterAsyncService {
    private final OkHttpClient httpClient = new OkHttpClient.Builder().build();
    private final RuleMatcher ruleMatcher;
    private final SqlConvert sqlConvert;
    private final ValueConfig valueConfig;
    private final FilterService filterService;
    private final ConcurrentHashMap<Integer, FilterProgress> progressMap;
    private final ConcurrentHashMap<Integer, KsRule> idKsRuleMap;
    private final BlockingQueue<BaseFilterParam> baseFilterParamList = new LinkedBlockingQueue<>();
    private final AtomicInteger filterRunningCount = new AtomicInteger(0);
    private final DateFormat dateFormat02 = new SimpleDateFormat("yyyy-MM-dd");

    private final DbUtil dbUtil;
    private final TaskServiceUtil taskServiceUtil;

    private final KsFilterRepository ksFilterRepository;

    private final JPAQueryFactory queryFactory;

    private final KsFilterResultService ksFilterResultService;

    public FilterAsyncService(RuleMatcher ruleMatcher, ValueConfig valueConfig, @Lazy FilterService filterService, DbUtil dbUtil, TaskServiceUtil taskServiceUtil, KsFilterRepository ksFilterRepository, JPAQueryFactory queryFactory, KsFilterResultService ksFilterResultService) {
        this.ruleMatcher = ruleMatcher;
        this.valueConfig = valueConfig;
        this.filterService = filterService;
        this.dbUtil = dbUtil;
        this.taskServiceUtil = taskServiceUtil;
        this.ksFilterRepository = ksFilterRepository;
        this.queryFactory = queryFactory;
        this.ksFilterResultService = ksFilterResultService;
        this.progressMap = new ConcurrentHashMap<>();
        idKsRuleMap = new ConcurrentHashMap<>();
        this.sqlConvert = new SqlConvert();
    }

    public FilterProgress getProgress(Integer filterId) {
        return progressMap.get(filterId);
    }

    public ConcurrentHashMap<Integer, FilterAsyncService.FilterProgress> getProgressMap() {
        return progressMap;
    }

    @Async("taskExecutor")
    public void filterSync(Integer recordId, FilterService.FilterBundle filterBundle) {
        runBaseFilter(true, recordId, filterBundle, null);
    }


    @Async("taskExecutor")
    public void filterAsync(Integer recordId, FilterService.FilterBundle filterBundle, JSONObject taskListParam) {
        runBaseFilter(false, recordId, filterBundle, taskListParam);
    }

    public void runBaseFilter(boolean isTask, Integer recordId, FilterService.FilterBundle filterBundle, JSONObject taskListParam) {
        baseFilterParamList.add(new BaseFilterParam(recordId, filterBundle, taskListParam));

        if (filterRunningCount.get() > 3) {
            return;
        }

        filterRunningCount.incrementAndGet();
        try {
            while (!baseFilterParamList.isEmpty()) {
                log.info("filter  ##start" + recordId);
                BaseFilterParam baseFilterParam = baseFilterParamList.poll();
                baseFilter(isTask, Objects.requireNonNull(baseFilterParam).recordId, baseFilterParam.filterBundle, baseFilterParam.taskListParam);
                log.info("filter  ##end" + recordId);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            filterRunningCount.getAndDecrement();
        }

    }

    public void baseFilter(boolean isTask, Integer recordId, FilterService.FilterBundle filterBundle, JSONObject taskListParam) {

        KsFilter ksFilter = queryFactory.selectFrom(qKsFilter).where(qKsFilter.id.eq(recordId)).fetchOne();
        if (ksFilter == null) {
            return;
        }

        if (!validTableNames(filterBundle, ksFilter)) {
            return;
        }

        // 添加进度
        progressMap.put(recordId, new FilterAsyncService.FilterProgress(recordId, 0, 0,
                0, filterBundle.getFilterParams().size()));

        try {
            filterBundle.getFilterParams().forEach(filterParam -> {
                try {
                    Integer ruleId = filterParam.getKsRuleId();
                    KsRule rule = getRuleById(ruleId);
                    if (rule == null || (rule.getIsAvailable() != null && rule.getIsAvailable() == 0)) {
                        return;
                    }

                    if (filterParam.getCondition() == null) {
                        filterParam.setCondition(new FilterService.FilterCondition());
                    }
                    String dbCode = filterParam.getDbCode();
                    if (StringUtils.isBlank(dbCode)) {
                        dbCode = queryFactory.select(qRlRuleTable.dbCode)
                                .from(qRlRuleTable)
                                .where(qRlRuleTable.tableName.eq(filterParam.getTableName()).
                                        and(qRlRuleTable.ruleId.eq(ruleId)))
                                .fetchFirst();
                    }


                    Map<String, SysSourceMappingTableField> ksTableFieldsMap = new HashMap<>();

                    BooleanExpression booleanExpression = qSysSourceMappingTableField.tableName.eq(filterParam.getTableName());

                    if (dbCode != null) {
                        booleanExpression = booleanExpression.and(qSysSourceMappingTableField.dbCode.eq(dbCode));
                    }

                    queryFactory.selectFrom(qSysSourceMappingTableField).where(booleanExpression).fetch().forEach(item -> ksTableFieldsMap.putIfAbsent(item.getTableField(), item));
                    List<SysSourceMappingTableField> ksTableFields = new ArrayList<>(ksTableFieldsMap.values());


                    String primaryField = "";
                    for (SysSourceMappingTableField tableField : ksTableFields) {
                        if (FieldConstants.PRIMARY_FIELD.equals(tableField.getTableFieldType())) {
                            primaryField = tableField.getTableField();
                            break;
                        }
                    }
                    final Map<String, String> fieldMap = ksTableFields
                            .stream()
                            .collect(Collectors.toMap(SysSourceMappingTableField::getTableField, SysSourceMappingTableField::getTableFieldName));
                    process(isTask, dbCode, taskListParam, ksFilter, filterBundle, rule, filterParam.getTableName(), fieldMap, filterParam, primaryField);
                } catch (Exception e) {
                    ksFilter.setError(e.getMessage());
                    e.printStackTrace();
                }
            });
            ksFilter.setStatus(1);
            ksFilterRepository.save(ksFilter);
        } catch (Exception e) {
            e.printStackTrace();
            ksFilter.setError(e.getMessage());
            ksFilter.setStatus(-1);
        } finally {
            if (progressMap.containsKey(recordId)) {
                progressMap.get(recordId).setRuleCurrent(
                        progressMap.get(recordId).getCurrent() + 1
                );
            }
            ksFilterRepository.save(ksFilter);

        }

        // 移除进度
        progressMap.remove(recordId);
        filterService.clueAutoResult(recordId);
    }

    public boolean validTableNames(FilterService.FilterBundle filterBundle, KsFilter ksFilter) {
        List<String> tableNames = filterBundle.getFilterParams()
                .stream()
                .map(FilterService.FilterParam::getTableName)
                .collect(Collectors.toList());

        List<String> ksTableList = queryFactory.select(qSysSourceMappingTable.tableName).from(qSysSourceMappingTable)
                .where(qSysSourceMappingTable.tableName.in(tableNames))
                .groupBy(qSysSourceMappingTable.tableName)
                .fetch();


        for (String tableName : tableNames) {
            if (!ksTableList.contains(tableName)) {
                ksFilter.setError(String.format("无效表 %s", tableName));
                ksFilter.setStatus(-1);
                ksFilterRepository.save(ksFilter);
                return false;
            }
        }
        return true;
    }

    private void process(boolean isTask, String dbCode, JSONObject taskListParam, KsFilter ksFilter, FilterService.FilterBundle filterBundle, KsRule ksRule, String table,
                         Map<String, String> fieldMap, FilterService.FilterParam param,
                         String primaryField) {


        String condition = createCondition(param.getCondition());

        FilterCbParam filterCbParam = new FilterCbParam();
        filterCbParam.setCondition(condition);
        filterCbParam.setTable(table);

        List<String> fields;
        if (valueConfig.getKsDbDriverClassName().equals("dm.jdbc.driver.DmDriver")) {
            fields = fieldMap.keySet()
                    .stream()
                    .map(s -> String.format("\"%s\" as \"%s\"", s, s))
                    .collect(Collectors.toList());
        } else {
            fields = fieldMap.keySet()
                    .stream()
                    .map(s -> String.format("`%s` as '%s'", s, s))
                    .collect(Collectors.toList());
        }

        filterCbParam.setFields(String.join(",", fields));

        if (dbCode == null) {
            dbCode = "172.16.137.24_supervise_platform";

        }
        filterCb(isTask, dbCode, taskListParam, ksFilter, ksRule, filterCbParam, primaryField, param);


        Integer match = filterBundle.getMatch();
        Integer total = 0;
        if (match == null) {
            match = 0;
        }
        for (FilterService.FilterParam inFp : filterBundle.getFilterParams()) {

            if (inFp.getTableName().equals(param.getTableName()) && inFp.getKsRuleId().equals(param.getKsRuleId())) {
                match += inFp.getMatch();
                total += inFp.getLastPosition();
            }
        }

        filterBundle.setMatch(match);
        filterBundle.setTotal(total);
        filterBundle.setNotMatch(total - match);
        // 更新信息
        ksFilter.setFilterBundle(JSONObject.toJSONString(filterBundle));

        ksFilterRepository.save(ksFilter);


    }


    private void filterCb(boolean isTask, String dbCode, JSONObject taskListParam, KsFilter ksFilterRecord, KsRule ksRule,
                          FilterCbParam filterCbParam, String primaryField, FilterService.FilterParam param) {

        Map<String, String> tableConditionMap = new HashMap<>();
        if (taskListParam != null) {
            tableConditionMap = taskListParam.getObject("tableConditionMap", new TypeReference<Map<String, String>>() {
            }.getType());
        }
        Integer recordId = ksFilterRecord.getId();
        if (isTask) {
            String dateFieldName = queryFactory.select(qKsTableField.tableField).from(qKsTableField).where(qKsTableField.tableName.eq(filterCbParam.getTable()).and(qKsTableField.tableFieldType.eq("date"))).fetchFirst();
            String dateCondition = filterCbParam.getCondition();
            if (StringUtils.isNotBlank(dateFieldName)) {
                Date date = new Date();
                Calendar calendar = Calendar.getInstance();
                calendar.setTime(date);
                calendar.add(Calendar.DAY_OF_YEAR, -13);
                if (!filterCbParam.getCondition().contains("WHERE")) {
                    dateCondition = String.format("WHERE %s >= '%s'", dateFieldName, dateFormat02.format(calendar.getTime()));
                } else {
                    dateCondition = filterCbParam.getCondition() + String.format("AND %s >= '%s'", dateFieldName, dateFormat02.format(calendar.getTime()));
                }
            }
            filterCbParam.setCondition(dateCondition);
        }

        if (filterCbParam.getCondition().trim().isEmpty()) {
            filterCbParam.setCondition(" WHERE 1=1 ");
        }

        List<String> resultIdList = new ArrayList<>();
        String dbCodeTable = dbCode + ":" + filterCbParam.getTable();
        String tableCondition = "";
        if (tableConditionMap != null && tableConditionMap.containsKey(dbCodeTable)) {
            tableCondition = tableConditionMap.get(dbCodeTable);
        }
        filterCbParam.setCondition(filterCbParam.getCondition() + tableCondition);

        int count = 0;

        if (ksRule.getRuleExpressionJson().contains("1#mysqlFunction")) {
            count = 1;
        } else {
            count = Integer.parseInt(String.valueOf(fetchSQL(dbCode, "SELECT COUNT(*) as count FROM " +
                    filterCbParam.getTable() + " " + filterCbParam.getCondition()).get(0).get("count")));
        }


        int batchSize = 10000;
        int threadSize = 10;
        if (count == 1) {
            threadSize = 1;
            batchSize = 1;
        }
        int threadBatchSize = batchSize / threadSize;

        int i = 0;

        BooleanExpression fieldExpression = qSysSourceMappingTableField.tableName.eq(filterCbParam.getTable()).and(qSysSourceMappingTableField.tableFieldType.eq(FieldConstants.PRIMARY_FIELD).and(qSysSourceMappingTableField.tableFieldDataType.contains("int")));
        if (dbCode == null) {
            fieldExpression = fieldExpression.and(qSysSourceMappingTableField.dbCode.isNull());
        } else {
            fieldExpression = fieldExpression.and(qSysSourceMappingTableField.dbCode.eq(dbCode));
        }

        String fields = filterCbParam.getFields();
        if (fields.trim().isEmpty()) {
            fields = "*";
        }
        AtomicInteger matchCount = new AtomicInteger(0);
        int lastPosition = 0;
        if (isTask) {
            lastPosition = param.getLastPosition();
        }
        final int[] current = {lastPosition};
        batchSize += lastPosition;
        while (current[0] < count) {
            int offset = batchSize * i;
            if (i == 0) {
                offset = lastPosition;
            }
            String resultQuery = "SELECT " + fields + " FROM "
                    + filterCbParam.getTable() + " " + filterCbParam.getCondition()
                    + " LIMIT " + offset + "," + batchSize;
            log.info("current:" + current[0] + ",count:" + count);

            i++;

            log.info("查询数据库 sql: " + resultQuery);
            List<Map<String, Object>> resultList = fetchSQL(dbCode, resultQuery);
            log.info("查询数据库结束 size: " + resultList.size());

            if (resultList.isEmpty()) {
                break;
            }

            List<SysSourceMappingTableField> sysSourceMappingTableFields = queryFactory.selectFrom(qSysSourceMappingTableField).where(qSysSourceMappingTableField.tableName.eq(filterCbParam.getTable()).and(qSysSourceMappingTableField.dbCode.eq(dbCode)).and(qSysSourceMappingTableField.ksDataFieldCode.isNotEmpty())).fetch();
            Map<String, String> fieldNameMap = sysSourceMappingTableFields.stream().collect(Collectors.toMap(SysSourceMappingTableField::getTableField, SysSourceMappingTableField::getKsDataFieldCode));
            AtomicInteger runCount = new AtomicInteger(0);
            List<String> updateList = new CopyOnWriteArrayList<>();
            ExecutorService pool = Executors.newCachedThreadPool();
            int finalCount = count;
            Map<String, String> finalTableConditionMap = tableConditionMap;
            Map<String, RlRuleTable> stringRlRuleTableMap = new HashMap<>();
            Map<String, String> systemCodePileUrlMap = new HashMap<>();


            boolean isLast = (resultList.size() % batchSize) != 0;
            int threadCount = isLast ? (resultList.size() % batchSize) / threadBatchSize + 1 : threadSize;
            CountDownLatch countDownLatch = new CountDownLatch(threadCount);
            Consumer<CountDownLatch> consumer = (c) -> {
                try {
                    List<Map<String, Object>> matchResultList = new ArrayList<>();
                    List<KsFilterResult> resultRecords = new ArrayList<>();
                    List<Map<String, Object>> recordList = resultList.stream().skip(runCount.getAndIncrement() * threadBatchSize).limit(threadBatchSize)
                            .collect(Collectors.toList());
                    for (Map<String, Object> record : recordList) {
                        fieldNameMap.forEach((key, value) -> {
                            String recordValue = String.valueOf(record.get(key));
                            record.put(value, recordValue);
                        });
                        // 匹配结果json包含结构
                        JSONObject cpJson = new JSONObject();
                        // 未匹配错误原因列表
                        List<String> errorList = new ArrayList<>();
                        List<String> tipsList = new ArrayList<>();
                        boolean match = filterItem(ksRule, record, cpJson, errorList, tipsList, dbCode);
                        KsFilterResult resultRecord = null;
                        try {
                            resultRecord = processCb(match, record, ksFilterRecord, ksRule, filterCbParam.table, primaryField, cpJson, errorList, tipsList);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

                        if (match) {
                            matchCount.getAndIncrement();
                            if (StringUtils.isNotBlank(ksRule.getAdminArea())) {
                                String systemCode = ksFilterRecord.getSystemCode();
                                String pileUrl;
                                if (systemCodePileUrlMap.containsKey(systemCode)) {
                                    pileUrl = systemCodePileUrlMap.get(systemCode);
                                } else {
                                    pileUrl = queryFactory.select(qSystemCodeUrl.url).from(qSystemCodeUrl).where(qSystemCodeUrl.systemCode.eq(systemCode)).fetchFirst();
                                    systemCodePileUrlMap.put(systemCode, pileUrl);
                                }
                                JSONObject jsonObject = new JSONObject();
                                jsonObject.put("name", ksRule.getRuleName());
                                jsonObject.put("content", ksRule.getDescription());
                                jsonObject.put("admin_area", ksRule.getAdminArea());
                                jsonObject.put("tags", record.get("tags"));
                                String sourceDataId = String.valueOf(record.get("id") == null ? record.get("ID") : record.get("id"));
                                jsonObject.put("sourceDataId", sourceDataId);
                                RequestBody body = RequestBody.create(MediaType.parse("application/json; charset=utf-8"), jsonObject.toJSONString());
                                Request request = new Request.Builder()
                                        .url(pileUrl + "/clue/entry")
                                        .post(body)
                                        .build();
                                try {
                                    Response response = httpClient.newCall(request).execute();
                                    response.close();
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                        RlRuleTable rlRuleTable;
                        String dbCodeTableNameRuleId = dbCodeTable + "-" + ksRule.getId();
                        if ((stringRlRuleTableMap.containsKey(dbCodeTableNameRuleId))) {
                            rlRuleTable = stringRlRuleTableMap.get(dbCodeTableNameRuleId);
                        } else {
                            rlRuleTable = queryFactory.selectFrom(qRlRuleTable).where(qRlRuleTable.ruleId.eq(ksRule.getId()).and(qRlRuleTable.tableName.eq(filterCbParam.getTable()).and(qRlRuleTable.dbCode.eq(dbCode)).and(
                                    (qRlRuleTable.matchContent.length().gt(0).and(qRlRuleTable.matchContent.ne("undefined"))).or(
                                            qRlRuleTable.noMatchContent.length().gt(0).and(qRlRuleTable.noMatchContent.ne("undefined"))
                                    )
                            ))).fetchFirst();
                            stringRlRuleTableMap.put(dbCodeTableNameRuleId, rlRuleTable);
                        }
                        if (rlRuleTable != null) {
                            String updateString;
                            if (match) {
                                updateString = "update " + rlRuleTable.getTableName() + " set " + rlRuleTable.getColumnName() + " = " + rlRuleTable.getMatchContent() + " where id = " + record.get("id");
                            } else {
                                updateString = "update " + rlRuleTable.getTableName() + " set " + rlRuleTable.getColumnName() + " = " + rlRuleTable.getNoMatchContent() + " where id = " + record.get("id");
                            }
                            updateList.add(updateString);
                        }
                        if (resultRecord != null && !resultRecord.getErrors().isEmpty()) {
                            int weight = RuleMatcher.getWeight();
                            resultRecord.setResultPower(ksFilterRecord.getResultPower());
                            resultRecord.setWeight(weight);
                            resultRecord.setDataSourceName(filterCbParam.getTable());
                            resultRecord.setDbCode(dbCode);
                            if (resultRecord.getDataId() == null || resultRecord.getDataId().equals("null")) {
                                resultRecord.setDataId(String.valueOf(System.currentTimeMillis()));
                            } else {
                                resultIdList.add(resultRecord.getDataId());
                            }
                            resultRecords.add(resultRecord);
                            matchResultList.add(record);
                        }
                    }
                    try {
                        if (taskListParam != null && finalTableConditionMap != null && !resultIdList.isEmpty()) {
                            finalTableConditionMap.put(dbCodeTable, " AND id in (" + StringUtils.join(resultIdList, ",") + ")");
                            taskListParam.put("tableConditionMap", finalTableConditionMap);
                            taskServiceUtil.executeTask(taskListParam);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                    if (!resultRecords.isEmpty()) {
                        resultRecords.forEach(item -> {
                                    try {
                                        ksFilterResultService.save(item);
                                    } catch (DataIntegrityViolationException e) {
                                        log.warning("重复id");
                                        e.printStackTrace();
                                    }
                                }
                        );
                        pushToClue(filterCbParam.getTable(), ksFilterRecord, ksRule, dbCode, matchResultList);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    c.countDown();
                    current[0] += threadBatchSize;
                    if (!isTask) {
                        progressCb(recordId, current[0], finalCount);
                    }
                }
            };
            for (int j = 0; j < threadCount; j++) {
                pool.execute(() -> consumer.accept(countDownLatch));
            }
            try {
                countDownLatch.await();
            } catch (Exception ignore) {
            }

            pool.shutdown();
            updateList.forEach(u -> {
                executeSQL(dbCode, u);
            });
        }
        int match = matchCount.get();

        param.setMatch(match);
        param.setLastPosition(count);
        param.setTotal(count - lastPosition);
        RuleMatcher.getRuleCache().clear();
        idKsRuleMap.clear();
    }

    private boolean filterItem(KsRule ksRule, Map<String, Object> dataMap,
                               JSONObject cpJson, List<String> errorList, List<String> tipsList, String dbCode) {
        if (ksRule != null) {
            JSONObject rootRuleExpressionJson = JSONObject.parseObject(String.valueOf(ksRule.getRuleExpressionJson()));
            try {
                return ruleMatcher.matchRuleByMap(dataMap, rootRuleExpressionJson, cpJson, errorList, tipsList, dbCode);
            } catch (Exception e) {
                log.warning(e.getMessage());
            }
        }
        return false;
    }

    public ResponseEntity<Object> pushToClue(String dataSourceName, KsFilter ksFilterRecord, KsRule rule, String dbCode, List<Map<String, Object>> sourceDataList) {
        if (ksFilterRecord.getPushClue() == null || !ksFilterRecord.getPushClue()) {
            return ResponseEntity.ok("");
        }
        String systemCode = ksFilterRecord.getSystemCode();
        if (systemCode == null) {
            return ResponseEntity.ok("");
        }
        String pileUrl = queryFactory.select(qSystemCodeUrl.url).from(qSystemCodeUrl).where(qSystemCodeUrl.systemCode.eq(systemCode)).fetchFirst();
        BooleanExpression booleanExpression = qSysSourceMappingTableField.tableName.eq(dataSourceName);
        if (dbCode != null) {
            booleanExpression = booleanExpression.and(qSysSourceMappingTableField.dbCode.eq(dbCode));
        }
        List<SysSourceMappingTableField> sourceFieldList = queryFactory.selectFrom(qSysSourceMappingTableField).where(booleanExpression).fetch();
        SysSourceMappingTableField leftPKRow = sourceFieldList.stream().filter(r -> ("primary").equals(r.getTableFieldType())).findAny().orElse(null);
        String leftPK = "";
        if (leftPKRow != null) {
            leftPK = leftPKRow.getTableField();
        }

        List<PileDataClue> pileDataClueList = new ArrayList<>();
        Map<String, String> colMap = new HashMap<>();

        final String CASE_NAME = "caseName";
        final String CONTENT = "content";
        final String SUMMARY = "summary";
        final String CREATE_TIME = "createTime";
        final String CASE_TIME = "caseTime";
        final String CASE_TYPE = "caseType";
        final String ADMIN_DEPT = "adminDept";
        final String BMSAH = "bmsah";
        final String KEYWORD = "keyword";
        final String TAG = "tag";
        final String RYBM = "rybm";
        final String DWBM = "dwbm";
        final String DEP_CODE = "dep_code";
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        sourceFieldList.forEach(field -> {
            String searchType = field.getTableFieldSearchType();
            String fieldName = field.getTableField();
            if (searchType != null) {
                switch (searchType) {
                    case "标题":
                        colMap.put(CASE_NAME, fieldName);
                        break;
                    case "内容":
                        colMap.put(CONTENT, fieldName);
                        break;
                    case "摘要":
                        colMap.put(SUMMARY, fieldName);
                        break;
                    case "录入时间":
                        colMap.put(CREATE_TIME, fieldName);
                        break;
                    case "案件时间":
                        colMap.put(CASE_TIME, fieldName);
                        break;
                    case "案件类型":
                        colMap.put(CASE_TYPE, fieldName);
                        break;
                    case "办案单位":
                        colMap.put(ADMIN_DEPT, fieldName);
                        break;
                    case "部门受案号":
                        colMap.put(BMSAH, fieldName);
                        break;
                    case "关键字":
                        colMap.put(KEYWORD, fieldName);
                        break;
                    case "标签":
                        colMap.put(TAG, fieldName);
                        break;
                    case "人员编码":
                        colMap.put(RYBM, fieldName);
                        break;
                    case "部门编码":
                        colMap.put(DEP_CODE, fieldName);
                        break;
                    case "单位编码":
                        colMap.put(DWBM, fieldName);
                        break;
                    default:
                        break;
                }
            }

        });

        String finalLeftPK = leftPK;
        sourceDataList.forEach(source -> {
            String sourcePower = ksFilterRecord.getPower();
            if (sourcePower == null) {
                sourcePower = "";
            }
            String sourceDataId;
            if (source.get(finalLeftPK) != null) {
                sourceDataId = String.valueOf(source.get(finalLeftPK));
            } else {
                sourceDataId = String.valueOf(System.currentTimeMillis());
            }
            StringBuilder power = new StringBuilder(sourcePower);
            PileDataClue clueRecord = new PileDataClue();
            clueRecord.setSourceDataId(sourceDataId);
            clueRecord.setProcessStatus("0");
            clueRecord.setIsDelete(false);
            clueRecord.setClueType(0);
            clueRecord.setCaseName("暂无");
            clueRecord.setTags(rule.getRuleName());
            clueRecord.setClueId(String.valueOf(UUID.randomUUID()));
            clueRecord.setCasebaseId(rule.getId());
            clueRecord.setIsMoveIn(1);
            clueRecord.setImportance(0);
            clueRecord.setClueSource("统一业务系统");
            clueRecord.setSourceDataType("案情");
            try {
                clueRecord.setTransferCase(String.valueOf(source.get("transferCase")));
            } catch (Exception e) {
                e.printStackTrace();
            }
            try {
                clueRecord.setRemarks(String.valueOf(source.get("remarks")));
            } catch (Exception e) {
                e.printStackTrace();
            }
            if (source.containsKey("id")) {
                clueRecord.setSourceDataId(String.valueOf(source.get("id")));
            }
            if (ksFilterRecord.getPushSearchTypePower() != null && ksFilterRecord.getPushSearchTypePower()) {
                if (colMap.get(DEP_CODE) != null) {
                    power.append(",depCode_").append(source.get(colMap.get(DEP_CODE)));
                }
                if (colMap.get(RYBM) != null) {
                    power.append(",rybm_").append(source.get(colMap.get(RYBM)));
                }
                if (colMap.get(DWBM) != null) {
                    power.append(",dwbm_").append(source.get(colMap.get(DWBM)));
                }
            }
            clueRecord.setPower(power.toString());
            if (colMap.get(CASE_NAME) != null) {
                clueRecord.setCaseName(String.valueOf(source.get(colMap.get(CASE_NAME))));
            }
            if (colMap.get(CONTENT) != null) {
                clueRecord.setContent(String.valueOf(source.get(colMap.get(CONTENT))));
            }
            if (colMap.get(SUMMARY) != null) {
                clueRecord.setSummary(String.valueOf(source.get(colMap.get(SUMMARY))));
                clueRecord.setContent(String.valueOf(source.get(colMap.get(SUMMARY))));
            }
            if (colMap.get(CREATE_TIME) != null) {
                try {
                    clueRecord.setCreationTime(Timestamp.from(sdf.parse(source.get(colMap.get(CREATE_TIME)).toString()).toInstant()));
                    clueRecord.setEarliestReportTime(Timestamp.from(sdf.parse(source.get(colMap.get(CREATE_TIME)).toString()).toInstant()));
                } catch (ParseException e) {
                    e.printStackTrace();
                }
            }
            if (colMap.get(CASE_TIME) != null) {
                try {
                    clueRecord.setEarliestReportTime(Timestamp.from(sdf.parse(source.get(colMap.get(CASE_TIME)).toString()).toInstant()));
                } catch (ParseException e) {
                    e.printStackTrace();
                }
            }
            if (colMap.get(CASE_TYPE) != null) {
                clueRecord.setCaseType(String.valueOf(source.get(colMap.get(CASE_TYPE))));
            }
            if (colMap.get(ADMIN_DEPT) != null) {
                clueRecord.setAdminDept(String.valueOf(source.get(colMap.get(ADMIN_DEPT))));
                clueRecord.setAdminArea(String.valueOf(source.get(colMap.get(ADMIN_DEPT))));
            }

            if (colMap.get(BMSAH) != null) {
                clueRecord.setSourceDataId(String.valueOf(source.get(colMap.get(BMSAH))));
            }
            if (colMap.get(KEYWORD) != null) {
                clueRecord.setKeywords(String.valueOf(source.get(colMap.get(KEYWORD))));
            }
            if (colMap.get(TAG) != null) {
                clueRecord.setTags(String.valueOf(source.get(colMap.get(TAG))));
            }

            pileDataClueList.add(clueRecord);
        });

        JSONObject param = new JSONObject();
        param.put("pileDataClueList", pileDataClueList);
        log.info("" + pileDataClueList.size());
        final OkHttpClient okHttpClient = new OkHttpClient();
        final MediaType JSON_TYPE = MediaType.parse("application/json;charset=utf-8");
        RequestBody body = RequestBody.create(JSON_TYPE, param.toJSONString());
        Request request = new Request.Builder()
                .url(pileUrl + "/clue/batchInsertClueFromCompareTaskResult")
                .post(body)
                .build();
        try {
            Response response = okHttpClient.newCall(request).execute();
            log.info(Objects.requireNonNull(response.body()).string());
            response.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ResponseEntity.ok().build();
    }


    public String createCondition(FilterService.FilterCondition filterCondition) {
        if (filterCondition == null) {
            return "WHERE 1=1";
        }
        JSONArray and = new JSONArray();

        for (FilterService.ClassCondition classCondition :
                Optional.ofNullable(filterCondition.getClassConditions()).orElse(new ArrayList<>())) {
            SqlConvert.SqlCondition sqlCondition = new SqlConvert.SqlCondition();
            sqlCondition.setLeftField(classCondition.getFieldName());
            if (classCondition.getValues().size() > 1) {
                String val = String.format("'%s'", String.join("','", classCondition.getValues()));
                sqlCondition.setRightField(val);
                sqlCondition.setOp(IN);
                and.add(JSONObject.parseObject(JSONObject.toJSONString(sqlCondition)));
            } else if (classCondition.getValues().size() == 1) {
                sqlCondition.setRightField(String.format("'%s'", classCondition.getValues().get(0)));
                sqlCondition.setOp("=");
                and.add(JSONObject.parseObject(JSONObject.toJSONString(sqlCondition)));
            }
        }

        if (!and.isEmpty()) {
            JSONObject json = new JSONObject();
            json.put(AND, and);
            Date date = new Date();
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(date);
            calendar.add(Calendar.DAY_OF_YEAR, -13);
            return String.format("WHERE %s ", sqlConvert.condition(json));
        } else {
            return "WHERE 1=1";
        }
    }

    private KsRule getRuleById(Integer id) {
        if (idKsRuleMap.containsKey(id)) {
            return idKsRuleMap.get(id);
        }
        KsRule ksRule = queryFactory.selectFrom(qKsRule).where(qKsRule.id.eq(id)).fetchFirst();
        idKsRuleMap.put(id, ksRule);
        return ksRule;
    }

    private void executeSQL(String dbCode, String sql) {
        dbUtil.execute(dbCode, sql);
    }

    private List<Map<String, Object>> fetchSQL(String dbCode, String sql) {
        return dbUtil.query(dbCode, sql);
    }

    public KsFilterResult processCb(boolean match, Map<String, Object> record, KsFilter ksFilter, KsRule rule, String tableName, String primaryField,
                                    JSONObject cpJson, List<String> errorList, List<String> tipsList) {
        KsFilterResult result = new KsFilterResult();
        result.setRuleId(rule.getId());
        result.setFilterId(ksFilter.getId());
        result.setResult(JSONObject.toJSONString(record));
        result.setSuccess(match);
        result.setCompareJson(cpJson.toJSONString());
        String id = String.valueOf(record.get("id"));
        result.setDataId(id);
        if (StringUtils.isBlank(id)) {
            result.setDataId(String.valueOf(record.get(primaryField)));
        }
        String errors = String.join(",", errorList);
        String tips = String.join(",", tipsList);
        if (!errors.isEmpty()) {
            result.setStatus(-1);
            errors = ruleMatcher.varToValue(errors, record);
        } else if (!tips.isEmpty()) {
            result.setStatus(1);
            tips = ruleMatcher.varToValue(tips, record);
        } else {
            result.setStatus(0);
        }
        result.setErrors(errors);
        result.setTips(tips);

        boolean isForward = !rule.getRuleExpressionJson().contains("error");
        if (match && isForward) {
            String bmsahValue = String.valueOf(record.getOrDefault("bmsah", ""));
//            String tableAlias = queryFactory.select(qSysSourceMappingTable.tableAlias).from(qSysSourceMappingTable).where(qSysSourceMappingTable.tableName.eq(tableName)).fetchFirst();
//            if (StringUtils.isBlank(tableAlias)) {
//                tableAlias = tableName;
//            }
            String msg = "案件：" + bmsahValue + "，命中位置：" + tableName + "，命中模型：" + rule.getRuleName() + "，模型描述：" + rule.getDescription() + ";";
            result.setErrors(msg);
        }
        return result;
    }

    public void progressCb(int recordId, int current, int count) {
        // 更新进度
        if (progressMap.containsKey(recordId)) {
            progressMap.get(recordId).setCurrent(current);
            progressMap.get(recordId).setCount(count);
        }
    }

    @Data
    public static class FilterCbParam {
        private String fields;
        private String table;
        private String condition;
    }

    @Data
    @AllArgsConstructor
    public static class FilterProgress {
        private int id;

        private int current;

        private int count;

        private int ruleCurrent;

        private int ruleCount;
    }

    @Data
    @AllArgsConstructor
    public static class BaseFilterParam {
        Integer recordId;
        FilterService.FilterBundle filterBundle;
        JSONObject taskListParam;
    }
}

